Storage Transfer ServiceがS3へアクセスするときのIP範囲の変更検知方法を考えてみた

import functions_framework
import json
import requests
from google.cloud import firestore
from datetime import datetime

# Firestoreコレクションとドキュメントの設定
COLLECTION_NAME = 'ip_ranges'
DOCUMENT_NAME = 'current'

# STSのIP範囲公開JSON URL
GCS_IP_RANGES_URL = 'https://www.gstatic.com/storage-transfer-service/ipranges.json'

def filter_data(data):
    """比較対象のデータから不要なフィールドを除外"""
    if data is None:
        return None
    return {
        "syncToken": data.get("syncToken"),
        "prefixes": data.get("prefixes")
    }

def find_differences(old_prefixes, new_prefixes):
    """新旧の prefixes を比較して差分を見つける"""
    old_set = set(json.dumps(prefix, sort_keys=True) for prefix in old_prefixes)
    new_set = set(json.dumps(prefix, sort_keys=True) for prefix in new_prefixes)

    added = new_set - old_set
    removed = old_set - new_set

    return {
        "added": [json.loads(item) for item in added],
        "removed": [json.loads(item) for item in removed]
    }

@functions_framework.http
def detect_gcs_ip_changes(request):
    try:
        # STSのIP範囲取得
        response = requests.get(GCS_IP_RANGES_URL)
        response.raise_for_status()
        new_ip_ranges = response.json()

        db = firestore.Client()
        doc_ref = db.collection(COLLECTION_NAME).document(DOCUMENT_NAME)

        # Firestoreから設定済みIP範囲取得
        doc_snapshot = doc_ref.get()
        if doc_snapshot.exists:
            old_ip_ranges = doc_snapshot.to_dict()
        else:
            old_ip_ranges = None

        # 差分があるかどうかをチェック
        if old_ip_ranges is None:
            print('No existing IP ranges found in Firestore. Saving the initial data.')

            # 初回データ保存
            new_ip_ranges['lastUpdated'] = datetime.utcnow().isoformat() + 'Z'
            doc_ref.set(new_ip_ranges)
            return 'Initial IP ranges saved.', 200

        # 差分比較
        differences = find_differences(
            old_prefixes=filter_data(old_ip_ranges).get("prefixes", []),
            new_prefixes=filter_data(new_ip_ranges).get("prefixes", [])
        )

        added = differences["added"]
        removed = differences["removed"]

        if added or removed:
            print('IP ranges have changed.')
            print(f'Added IPs: {json.dumps(added, indent=2)}')
            print(f'Removed IPs: {json.dumps(removed, indent=2)}')

            # Firestoreは更新せずログ出力のみ実施
            return 'IP ranges have changed. Check logs for details.', 200
        else:
            print('No changes in IP ranges.')
            return 'No changes in IP ranges.', 200

    except requests.exceptions.RequestException as e:
        print(f'Error fetching IP ranges: {e}')
        return 'Error fetching IP ranges.', 500

    except Exception as e:
        print(f'Error: {e}')
        return 'Error detecting IP changes.', 500

functions-framework==3.*
requests
google-cloud-firestore

curl -m 70 -X POST "Cloud Run関数のURL" -H "Authorization: bearer $(gcloud auth print-identity-token)"

Initial IP ranges saved.

IP ranges have changed. Check logs for details.